package cn.getech.data.development.service.impl;

import cn.getech.data.development.config.properties.BdpJobConfig;
import cn.getech.data.development.config.properties.OozieConfig;
import cn.getech.data.development.constant.*;
import cn.getech.data.development.dto.JobInfoDto;
import cn.getech.data.development.entity.*;
import cn.getech.data.development.mapper.*;
import cn.getech.data.development.model.vo.JobInfoDevOpsVO;
import cn.getech.data.development.model.vo.SumOverViewVO;
import cn.getech.data.development.service.JobInfoService;
import cn.getech.data.development.service.JobNodeInfoService;
import cn.getech.data.development.service.JobRunHistoryService;
import cn.getech.data.development.service.ScheduleTimeService;
import cn.getech.data.development.utils.DataPermissionUtil;
import cn.getech.data.development.utils.HdfsUtil;
import cn.getech.data.development.utils.oozie.CoordinatorUtil;
import cn.getech.data.development.utils.oozie.JobUtil;
import cn.getech.data.intelligence.common.exception.RRException;
import cn.getech.data.intelligence.common.utils.PageUtils;
import cn.getech.data.intelligence.common.utils.Query;
import cn.getech.system.center.model.dto.SysUserDto;
import cn.getech.system.center.utils.ShiroUtils;
import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import jodd.util.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.oozie.client.OozieClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;

/**
 * <p>
 * 任务信息表 服务实现类
 * </p>
 *
 * @author zenith
 * @since 2018-08-23
 */
@Slf4j
@Service
public class JobInfoServiceImpl extends ServiceImpl<JobInfoMapper, JobInfo> implements JobInfoService {

    @Resource
    private JobNodeInfoMapper jobNodeInfoMapper;

    @Resource
    private ProcInfoMapper procInfoMapper;

    @Resource
    private JobNodeConfMapper jobNodeConfMapper;
    @Resource
    private JobNodeRunHistoryMapper jobNodeRunHistoryMapper;
    @Resource
    private JobInfoMapper jobInfoMapper;
    @Resource
    private JobRunHistoryMapper jobRunHistoryMapper;
    @Autowired
    private JobNodeInfoService jobNodeInfoService;
    @Resource
    private BdpJobConfig bdpJobConfig;
    @Autowired
    private OozieConfig oozieConfig;
    @Autowired
    private JobRunHistoryService jobRunHistoryService;

    @Autowired
    private JobDepRunHistoryMapper jobDepRunHistoryMapper;
    @Autowired
    private WorkMenuJobRelMapper workMenuJobRelMapper;
    @Autowired
    private WorkFlowDepMapper workFlowDepMapper;
    @Autowired
    private ScheduleTimeService scheduleTimeService;

    @Override
    public PageUtils queryPage(Map<String, Object> params) {
        String name = (String) params.get("name");
        String proc_id = (String) params.get("proc_id");
        String state = (String) params.get("state");
        String user = (String) params.get("user_id");//0：所有人，1：自己
        Integer enable = -1;
        Integer userId = -1;
        if (StringUtils.isNotBlank(state)) {
            enable = Integer.parseInt(state);
        }
        if (StringUtils.isNotBlank(user)) {
            userId = Integer.parseInt(user);
        }
        IPage<JobInfo> page = this.page(
                new Query<JobInfo>().getPage(params),
                new QueryWrapper<JobInfo>().like(StringUtils.isNotBlank(name), "name", name)
                        .eq(StringUtils.isNotBlank(proc_id), "proc_id", proc_id)
                        .eq(StringUtils.isNotBlank(state), "enable", enable)
                        .eq(StringUtils.isNotBlank(user), "create_per", userId)
        );

        return new PageUtils(page);
    }

    @Override
    public boolean checkIfRef(Integer id) {
        return jobNodeInfoMapper.selectCount(new QueryWrapper<JobNodeInfo>().eq("job_id", id).ne("type_id", 0)) > 0;
    }

    @Override
    public boolean checkExistById(Integer id) {
        return this.baseMapper.selectById(id) != null;
    }

    @Override
    public boolean checkPermission(Integer id) {
        JobInfo entity = this.baseMapper.selectById(id);
        return entity.getCreatePer() == ShiroUtils.getUserId().intValue();
    }

    @Override
    public boolean checkIsRun(Integer id) {
        JobInfo jobInfo = this.baseMapper.selectById(id);
        return (jobInfo.getLastRunState() != null) && LastRunState.RUNNING.getCode() == jobInfo.getLastRunState();
    }

    @Override
    public boolean checkExistByNameAndProcId(String name, Integer procId) {
        return this.baseMapper.selectCount(new QueryWrapper<JobInfo>().eq("name", name).eq("proc_id", procId)) > 0;
    }

    @Override
    public boolean checkJobIsEable(Integer jobId) {
        JobInfo entity = this.baseMapper.selectById(jobId);
        if (null == entity) {
            return false;
        }
        return !entity.getEnable();
    }


    /**
     * 封装删除所有此job相关的信息:删除jobInfo,jenkins,jobRunHistory,jobNodeRunHistory,
     *
     * @param jobInfoId
     */
    @Override
    public void deleteAllOfJob(Integer jobInfoId, Long userId) throws Exception {
        List<JobNodeInfo> jobNodeInfoList = jobNodeInfoMapper.selectList(new QueryWrapper<JobNodeInfo>().eq("job_id", jobInfoId));
        HdfsUtil hdfsUtil = new HdfsUtil(bdpJobConfig);
        OozieClient wc = new OozieClient(oozieConfig.getUrl());
        JobUtil jobUtil = new JobUtil(hdfsUtil, wc);
        //删除hdfs的相关job的workflow与shell文件
        String oozieJobId = this.getById(jobInfoId).getLastOozieJobId();
        jobUtil.deleteJob(userId, jobInfoId,oozieJobId);
        if(null != hdfsUtil){
            hdfsUtil.close();
        }
        if(CollectionUtil.isNotEmpty(jobNodeInfoList)){
            for (JobNodeInfo j : jobNodeInfoList) {
                //删除节点相关
                jobNodeConfMapper.delete(new QueryWrapper<JobNodeConf>().eq(StringUtil.isNotBlank(String.valueOf(j.getId())), "job_node_id", j.getId()).eq("job_type", 0));//根据job_node_id删除节点配置表
                jobNodeRunHistoryMapper.delete(new QueryWrapper<JobNodeRunHistory>().eq(StringUtil.isNotBlank(String.valueOf(j.getId())), "job_node_info_id", j.getId()));//根据job_node_id删除job_node_run_history
            }
        }

        jobNodeInfoMapper.delete(new QueryWrapper<JobNodeInfo>().eq(StringUtil.isNotBlank(String.valueOf(jobInfoId)), "job_id", jobInfoId));//最后删除jobnodeinfo中的节点信息
        //删除任务相关

        try {
            jobInfoMapper.deleteById(jobInfoId);//删除任务表
            WorkMenuJobRel workMenuJobRel = workMenuJobRelMapper.selectOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", WorkFlowType.WORKFLOW.getCode()).eq("work_flow_id", jobInfoId));
            jobNodeConfMapper.delete(new QueryWrapper<JobNodeConf>().eq(StringUtil.isNotBlank(String.valueOf(workMenuJobRel.getWorkMenuId())), "job_node_id", workMenuJobRel.getWorkMenuId()).eq("job_type", 1));//根据job_node_id删除节点配置表
            jobRunHistoryMapper.delete(new QueryWrapper<JobRunHistory>().eq(StringUtil.isNotBlank(String.valueOf(jobInfoId)), "job_info_id", jobInfoId));
        } catch (Exception e) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOB_DELETE.getMessage());
        }
    }

    @Override
    public void stopJobInfo(Integer jobId, String oozieJobId) throws Exception {
        //先停止其下的所有的任务和禁止任务---这里的job需要按照对应的连接排序的顺序
        List<JobNodeInfo> listJobNodeInfoList = jobNodeInfoService.selectJobNodeInfoByJobId(jobId);

        if (CollectionUtil.isEmpty(listJobNodeInfoList)) {
            return;
        }
        if (listJobNodeInfoList.size() <= 2) {
            return;
        }
        JobInfo jobInfo = jobInfoMapper.selectById(jobId);
        if (null == jobInfo) {
            return;
        }

        //kill掉oozie中的job
        OozieClient wc = new OozieClient(oozieConfig.getUrl());
        HdfsUtil hdfsUtil = new HdfsUtil(bdpJobConfig);
        JobUtil jobUtil = new JobUtil(hdfsUtil, wc);
        jobUtil.stopJob(oozieJobId);
        if(null != hdfsUtil){
            hdfsUtil.close();
        }

        //开始节点和结束节点不用停止
        List<JobNodeInfo> newListJobNodeInfoList = new ArrayList<>();
        for (int i = 1; i < listJobNodeInfoList.size() - 1; i++) {
            newListJobNodeInfoList.add(listJobNodeInfoList.get(i));
        }

        //并将运行中的状态全部转变为失败
        changeRunningJobNode(jobId);
    }

    @Override
    public List<SysUserDto> getJobOwnersList(Integer procId) {
        return this.jobInfoMapper.getJobOwnersList(procId);

    }

    @Override
    public List<JobInfoDto> selectListByIds(Set<Integer> jobIds) {
        return this.jobInfoMapper.selectListByIds(jobIds);

    }

    @Override
    public List<JobInfoDto> listAllNodeList(Set<Integer> nodeIds) {
        if (CollectionUtil.isEmpty(nodeIds)) {
            return null;
        }
        List<JobInfoDto> jobInfoDtos = jobInfoMapper.selectListByIds1(nodeIds);
        if (CollectionUtil.isEmpty(jobInfoDtos)) {
            return null;
        }
        //查询到对应的节点的配置信息
        for (JobInfoDto jobInfoDto : jobInfoDtos) {
//            List<JobNodeConf> jobNodeConfs = jobNodeConfMapper.selectList(new QueryWrapper<JobNodeConf>().eq("job_type", 1).eq("job_node_id", jobInfoDto.getId()));
            jobInfoDto.setWorkFlowNodeName("mid_" + jobInfoDto.getId());
//            if(CollectionUtil.isNotEmpty(jobNodeConfs)){
//                jobInfoDto.setJobNodeConfig(JobNodeConfig.listToJobNodeConfig(jobNodeConfs));
//            }
        }
        return jobInfoDtos;
    }

    @Override
    public void stopAllJobAndJobNodeHistory(Integer jobId, String lastOozieJobId, String lastCorOozieJobId) {
        //改变状态
        updateStatus(jobId);

        //停止手动任务
        if (StringUtils.isNotEmpty(lastOozieJobId)) {
            stopJob(jobId,lastOozieJobId);
        }

        //停止定时任务
        stopAutoJob(jobId);
    }
    @Override
    public void stopJob(Integer jobId, String lastOozieJobId) {
        //停止手动任务
        if (StringUtils.isNotEmpty(lastOozieJobId)) {
            //改变其历史中的记录
            List<JobRunHistory> jobRunHistoryList = jobRunHistoryMapper.selectList(new QueryWrapper<JobRunHistory>()
                    .eq("job_info_id", jobId)
                    .eq("state", LastRunState.RUNNING.getCode())
                    .eq("oozie_job_id", lastOozieJobId));
            if (CollectionUtil.isNotEmpty(jobRunHistoryList)) {
                jobRunHistoryMapper.batchUpdateStatus(jobRunHistoryList);
            }
            //其对应的节点的运行状态
            List<JobNodeRunHistory> jobNodeRunHistories = jobNodeRunHistoryMapper.selectList(new QueryWrapper<JobNodeRunHistory>()
                    .eq("job_info_id", jobId)
                    .eq("state", LastRunState.RUNNING.getCode())
                    .eq("oozie_job_id", lastOozieJobId));
            if (CollectionUtil.isNotEmpty(jobNodeRunHistories)) {
                for (JobNodeRunHistory jobNodeRunHistory : jobNodeRunHistories) {
                    //改变其log中的运行状态
                    if (null != jobNodeRunHistory) {
                        String logs = jobNodeRunHistory.getLog();
                        if (StringUtils.isNotEmpty(logs)) {
                            try {
                                JSONObject obj = JSONObject.parseObject(logs);
                                if (null != obj) {
                                    String status = obj.getString("status");
                                    if (StringUtils.isNotEmpty(status) && status.trim().toLowerCase().equals("running")) {
                                        obj.put("status", "killed");
                                        jobNodeRunHistory.setLog(obj.toJSONString());
                                    }
                                }
                            } catch (Exception e) {
                                log.error("转换日志异常！error:{}", e.getMessage());
                            }
                        }
                    }
                }
                jobNodeRunHistoryMapper.batchUpdateStatus(jobNodeRunHistories);
            }
        }
    }


    @Override
    public List<Integer> selectJobIdsByProcId(Integer procId) {
        List<JobInfo> jobInfos = jobInfoMapper.selectList(new QueryWrapper<JobInfo>().eq("proc_id", procId));
        List<Integer> jobInfoIds = new ArrayList<>();
        if (CollectionUtil.isNotEmpty(jobInfos)) {
            jobInfos.forEach(jobInfo -> {
                jobInfoIds.add(jobInfo.getId());
            });
        }
        return jobInfoIds;
    }

    @Override
    public List<Integer> selectJobIds() {
        Long userId = ShiroUtils.getUserId();
        Integer tenantId = ShiroUtils.getTenantId();
        if (DataPermissionUtil.isAdmin(userId.intValue())) {
            userId = null;
        }
        List<ProcInfo> authorisedProds = procInfoMapper.listUserProcs(userId, tenantId);
        List<Integer> authorisedProdIds = authorisedProds.stream().map(ProcInfo::getId).collect(Collectors.toList());
        QueryWrapper<JobInfo> query = new QueryWrapper<>();
//        query.and(w -> w.in("proc_id", authorisedProdIds));
        authorisedProdIds.add(-1);
        query.in("proc_id", new ArrayList<>(authorisedProdIds));
        List<JobInfo> jobInfos = jobInfoMapper.selectList(query);
        List<Integer> jobInfoIds = new ArrayList<>();
        if (CollectionUtil.isNotEmpty(jobInfos)) {
            jobInfos.forEach(jobInfo -> {
                jobInfoIds.add(jobInfo.getId());
            });
        }
        return jobInfoIds;
    }

    @Override
    public List<Integer> selectProcIds() {
        Long userId = ShiroUtils.getUserId();
        Integer tenantId = ShiroUtils.getTenantId();
        if (DataPermissionUtil.isAdmin(userId.intValue())) {
            userId = null;
        }
        List<ProcInfo> authorisedProds = procInfoMapper.listUserProcs(userId, tenantId);
        List<Integer> procIds = new ArrayList<>();
        if(CollectionUtil.isNotEmpty(authorisedProds)){
            procIds.addAll(authorisedProds.stream().map(ProcInfo::getId).distinct().collect(Collectors.toList()));
        }else{
            procIds.add(-1);
        }
        return procIds;
    }

    @Override
    public SumOverViewVO selectSumOverView() {
        return jobInfoMapper.selectOverView();
    }

    @Override
    public List<JobInfoDevOpsVO> changeBeans(List<WorkMenu> list) {
        if(CollectionUtil.isEmpty(list)){
            return null;
        }
        List<JobInfoDevOpsVO> restList = new ArrayList<>();
        List<ProcInfo> procList = procInfoMapper.listUserProcs(ShiroUtils.getUserId(), ShiroUtils.getTenantId());
        Set<Integer> procIds = new HashSet<>();
        if(CollectionUtil.isNotEmpty(procList)){
            procIds.addAll(procList.stream().map(ProcInfo::getId).collect(Collectors.toSet()));
        }
        //优化一下。改为批量查询
        List<Long> jobIds = new ArrayList<>();
        List<Long> depIds = new ArrayList<>();
        if(CollectionUtil.isNotEmpty(list)){
            list.forEach(o -> {
                if(Objects.equals(WorkFlowMenuType.WORKFLOW_FIELD.getCode(),o.getWorkType())){
                    jobIds.add(o.getId());
                }else{
                    depIds.add(o.getId());
                }
            });
            //查询所有的数据
            if(CollectionUtil.isNotEmpty(jobIds)){
                List<JobInfoDevOpsVO> ddd01s = jobInfoMapper.listJobInfoData(jobIds);
                if(CollectionUtil.isNotEmpty(ddd01s)){
                    restList.addAll(ddd01s);
                }
            }
            if(CollectionUtil.isNotEmpty(depIds)){
                List<JobInfoDevOpsVO> ddd02s = workFlowDepMapper.listJobInfoData(depIds);
                if(CollectionUtil.isNotEmpty(ddd02s)){
                    restList.addAll(ddd02s);
                }
            }
        }
        //转变数据格式
        if(CollectionUtil.isNotEmpty(restList)){
            for (JobInfoDevOpsVO jobInfoDevOpsVO : restList) {
                //最近一次运行的类型 -1：定时调度、-2：循环调度、id：手动操作的人id
                if(Objects.equals(-1,jobInfoDevOpsVO.getLastRunType())){
                    jobInfoDevOpsVO.setLastRunPerName("定时调度");
                }
                if(Objects.equals(-2,jobInfoDevOpsVO.getLastRunType())){
                    jobInfoDevOpsVO.setLastRunPerName("循环调度");
                }
                if(Objects.equals(-3,jobInfoDevOpsVO.getLastRunType())){
                    jobInfoDevOpsVO.setLastRunPerName("cronTab调度");
                }
                if(Objects.equals(jobInfoDevOpsVO.getScheduleType(),1)){
                    if(Objects.equals(2,jobInfoDevOpsVO.getCronType())){
                        //crontab表达式
                        if(StringUtils.isNotEmpty(jobInfoDevOpsVO.getCronUrl())){
                            jobInfoDevOpsVO.setNextLastRunTime(scheduleTimeService.cronTabNextDate(jobInfoDevOpsVO.getCronLastRunTime(),jobInfoDevOpsVO.getCronUrl()));
                        }
                    }else{
                        if(StringUtils.isNotEmpty(jobInfoDevOpsVO.getExpressionCron())){
                            if(Objects.equals(1,jobInfoDevOpsVO.getLastRunNum()) && Objects.equals(ScheduleRepeatType.INTERVAL.getCode(),jobInfoDevOpsVO.getScheduleOrInterval())){
                                jobInfoDevOpsVO.setNextLastRunTime(jobInfoDevOpsVO.getCronLastRunTime());
                            }else{
                                jobInfoDevOpsVO.setNextLastRunTime(changeExpressionCron(jobInfoDevOpsVO.getScheduleOrInterval(),jobInfoDevOpsVO.getExpressionCron(),jobInfoDevOpsVO.getCronLastRunTime()));
                            }
                        }
                    }
                }
                //判断是作业流还是作业流依赖
                if(!Objects.equals(jobInfoDevOpsVO.getWorkType(),3)){
                    //作业流依赖-禁用状态
                    if(Objects.equals(0,jobInfoDevOpsVO.getJobLayStatus())){
                        jobInfoDevOpsVO.setNextLastRunTime(null);
                        jobInfoDevOpsVO.setScheduleType(0);
                    }
                }
                WorkMenu workMenu = list.stream().filter(o -> Objects.equals(o.getId(), jobInfoDevOpsVO.getWorkFlowId())).findAny().orElse(null);
                if(null != workMenu){
                    if(null != jobInfoDevOpsVO){
                        if(procIds.contains(workMenu.getProcId())){
                            jobInfoDevOpsVO.setIsOwer(1);
                        }
                    }
                    //是否超时
                    if(null != jobInfoDevOpsVO){
                        jobInfoDevOpsVO.setIsOutHours(workMenu.getOutHours());
                    }
                }
            }
            //按照lastRunTime降序和id升序排序
            Collections.sort(restList,Comparator.comparing(JobInfoDevOpsVO::getLastRunTime,Comparator.nullsFirst(Date::compareTo)).reversed()
                    .thenComparing(JobInfoDevOpsVO::getWorkFlowId,Comparator.nullsLast(Long::compareTo)));
        }
        return restList;
    }

    private Date changeExpressionCron(Integer scheduleOrInterval, String expressionCron, Date lastRunTime) {
        if(StringUtils.isNotEmpty(expressionCron)){
            String crons = CoordinatorUtil.expressToCron(expressionCron, String.valueOf(scheduleOrInterval));
            return scheduleTimeService.getCronNextDateAll(lastRunTime, crons, scheduleOrInterval);
        }
        return null;
    }

    /**
     * 批量修改
     *
     * @param ids
     */
    @Override
    public void updateBatch(List<Long> ids,Integer userId) {
        jobInfoMapper.updateBatch(ids,userId);
    }

    @Override
    public void stopAutoJob(Integer jobId) {
        OozieClient wc = new OozieClient(oozieConfig.getUrl());
        HdfsUtil hdfsUtil = null;
        try {
            hdfsUtil = new HdfsUtil(bdpJobConfig);
        } catch (Exception e) {
            e.printStackTrace();
            if(hdfsUtil != null){
                hdfsUtil.close();
            }
            return;
        }
        JobUtil jobUtil = new JobUtil(hdfsUtil, wc);
        //停止定时任务
        List<JobRunHistory> jobRunHistoryList = jobRunHistoryMapper.selectRunningListLimit3(jobId);
        if (CollectionUtil.isNotEmpty(jobRunHistoryList)) {
            jobRunHistoryMapper.batchUpdateStatus(jobRunHistoryList);
            for (JobRunHistory jobRunHistory : jobRunHistoryList) {
                try {
                    jobUtil.stopJob(jobRunHistory.getOozieJobId());
                } catch (Exception e) {
                    log.error("暂停任务失败！error:{}", e.getMessage());
                }
                List<JobNodeRunHistory> jobNodeRunHistories = jobNodeRunHistoryMapper.selectList(new QueryWrapper<JobNodeRunHistory>()
                        .eq("oozie_job_id", jobRunHistory.getOozieJobId())
                        .in("state", LastRunState.RUNNING.getCode(),LastRunState.NOTFINISH.getCode()));
                if (CollectionUtil.isNotEmpty(jobNodeRunHistories)) {
                    for (JobNodeRunHistory jobNodeRunHistory : jobNodeRunHistories) {
                        //改变其log中的运行状态
                        if (null != jobNodeRunHistory) {
                            String logs = jobNodeRunHistory.getLog();
                            if (StringUtils.isNotEmpty(logs)) {
                                try {
                                    JSONObject obj = JSONObject.parseObject(logs);
                                    if (null != obj) {
                                        String status = obj.getString("status");
                                        if (StringUtils.isNotEmpty(status) && status.trim().toLowerCase().equals("running")) {
                                            obj.put("status", "killed");
                                            jobNodeRunHistory.setLog(obj.toJSONString());
                                        }
                                    }
                                } catch (Exception e) {
                                    log.error("转换日志异常！error:{}", e.getMessage());
                                }
                            }
                        }
                    }
                    jobNodeRunHistoryMapper.batchUpdateStatus(jobNodeRunHistories);
                }
            }
        }

        if(hdfsUtil != null){
            hdfsUtil.close();
        }
    }

    @Override
    public void updateStatus(Integer jobInfoId) {
        JobInfo jobInfo = jobInfoMapper.selectById(jobInfoId);
        if (null == jobInfo) {
            return;
        }
        if (Objects.equals(LastRunState.RUNNING.getCode(), jobInfo.getLastRunState())) {
            jobInfo.setLastRunState(LastRunState.STOP.getCode());
            if (null == jobInfo.getFailCount()) {
                jobInfo.setFailCount(1L);
            } else {
                jobInfo.setFailCount(jobInfo.getFailCount() + 1);
            }

            jobInfoMapper.updateById(jobInfo);
        }
    }

    /**
     * @Description 改变运行中的job为失败
     * @Author hujz
     * @Date 2019/12/25 15:03
     * @Param jobId
     * @Return
     * @Exception
     */
    @Transactional(rollbackFor = Exception.class)
    public void changeRunningJobNode(Integer jobId) {
        JobInfo jobInfo = jobInfoMapper.selectById(jobId);
        if (null == jobInfo) {
            return;
        }
        if (Objects.equals(LastRunState.RUNNING.getCode(), jobInfo.getLastRunState())) {
            jobInfo.setLastRunState(LastRunState.FAIL.getCode());
            jobInfo.setFailCount(jobInfo.getFailCount() + 1);
            jobInfoMapper.updateById(jobInfo);
        }
        //改变其历史中的记录
        List<JobRunHistory> jobRunHistoryList = jobRunHistoryMapper.selctListByIdLimit(10, jobId);
        if (CollectionUtil.isNotEmpty(jobRunHistoryList)) {
            for (JobRunHistory jobRunHistory : jobRunHistoryList) {
                if (Objects.equals(LastRunState.RUNNING.getCode(), jobRunHistory.getState())) {
                    jobRunHistory.setState(LastRunState.FAIL.getCode());
                    jobRunHistoryMapper.updateById(jobRunHistory);
                }
            }
        }
        //改变其节点的状态
        List<JobNodeInfo> listJobNodeInfoList = jobNodeInfoMapper.selectList(new QueryWrapper<JobNodeInfo>().eq("job_id", jobId));
        if (CollectionUtil.isEmpty(listJobNodeInfoList)) {
            return;
        }
        for (JobNodeInfo jobNodeInfo : listJobNodeInfoList) {
            List<JobNodeRunHistory> jobNodeRunHistoryList = jobNodeRunHistoryMapper.selctListByNodeIdLimit(10, jobId, jobNodeInfo.getNodeKey(), null);
            if (CollectionUtil.isNotEmpty(jobNodeRunHistoryList)) {
                for (JobNodeRunHistory jobNodeRunHistory : jobNodeRunHistoryList) {
                    if (Objects.equals(LastRunState.RUNNING.getCode(), jobNodeRunHistory.getState())) {
                        jobNodeRunHistory.setState(LastRunState.FAIL.getCode());
                        jobNodeRunHistoryMapper.updateById(jobNodeRunHistory);
                    }
                }
            }
        }
    }

}
